/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.threadpool;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Properties;
import java.util.Random;
import java.util.StringTokenizer;

import cgl.narada.transport.Link;
import cgl.narada.transport.LinkFactory;
import cgl.narada.transport.TransportDebugFlags;
import cgl.narada.transport.TransportException;
import cgl.narada.transport.TransportHandler;

/** This is a TCP thread pool based implementation of the LinkFactory 
    Interface. Lets assume that the connection is being initiated by a node A
    to another node B. There are two independent factors that need to be borne
    in mind while using LinkFactory. The first is of course setting the link 
    factory properties. The second is specifying the properties to create a 
    link to some other node B.
    
    <p>
    To initialize the PoolTCPLinkFactory one needs to specify the 
    <b>PoolTCPServerPort
    </b>. When the value of this variable is set to <b>0</b> it implies that 
    the node A initiating a connection to node B will not accept link creation
    requests from any other node.
    <p>
    To create a link to node B the following properties needs to be specified.
    <li><b>hostname</b>: This is the hostname on which the node B's process is
    running.
    <li><b>portnum</b>: This is the port number on which node B accepts link 
    creation requests from other nodes. In other words, the 
    <i>PoolTCPServerPort</i> specified to the PoolTCPChannelFactory at node B 
    is equal to portnum.
    
    @author Shrideep Pallickara
    $Date$
    $Revision$
*/

public class PoolTCPLinkFactory extends Thread 
  implements LinkFactory, TransportDebugFlags {

  private String managedLinkType = "pooltcp";
  private Properties properties;
  private String moduleName="PoolTCPLinkFactory: ";
  private int pooltcpServerSocketPort=0;
  private PoolTCPServerThread pooltcpServerThread;

  private Hashtable links;
  private TransportHandler transportHandler;
  
  private ThreadPoolManager threadPoolManager;
  private Selector poolSelector;
  
  private boolean keepLooping = true;
  private Hashtable poolLinks;

  private ServerSocketChannel ssc;
  
  private boolean registerChannel = false;
  private Random random;
  
  public PoolTCPLinkFactory() {
    links = new Hashtable();
    poolLinks = new Hashtable();
    random = new Random();
    try {
      poolSelector = Selector.open();
    } catch (Exception e) {
      System.out.println(moduleName + "Problems initializing selector " + e);
    }
  }
  
  
  /* These are the set of properties that are needed to set up the 
     link Factory. */
  public void setLinkFactoryProperties(Properties factoryProperties) {
    if (factoryProperties == null) {
      System.out.println(moduleName + "The specified properties is NULL");
      return;
    }
    properties = factoryProperties;
    String _valueString = properties.getProperty("PoolTCPServerPort");
    int _numWorkerThreads = 2;
    if (_valueString != null) { 
      pooltcpServerSocketPort = Integer.parseInt(_valueString);
      if (pooltcpServerSocketPort != 0) {
	_numWorkerThreads = 50;
      }
      threadPoolManager = new ThreadPoolManager(_numWorkerThreads);      
    } else {
      threadPoolManager = new ThreadPoolManager(_numWorkerThreads);
      System.out.println(moduleName + "Property PoolTCPServerPort missing!");
    }

    System.out.println(moduleName + "Initialized thread pool of [" +
			 _numWorkerThreads + "] worker threads");
  }
  
  
  /** Return the type of all the links created/managed by this factory */
  public String getManagedLinksType() {
    return managedLinkType;
  }
  
  /** Create a communication link based on the properties that have been
      specified. */
  public Link createLink(Properties linkProperties) 
    throws TransportException {
    if (linkProperties == null) {
      throw new TransportException(moduleName + 
				   "The specified link properties is null");
    }
    
    String _hostName = linkProperties.getProperty("hostname");
    String _portString = linkProperties.getProperty("portnum");
    
    if (_hostName == null || _portString == null) {
      throw new TransportException(moduleName + 
				   "One of the specified properties is null");
    }
    int _portNum = Integer.parseInt(_portString);
    
    try {
      InetSocketAddress _ia = new InetSocketAddress(_hostName, _portNum);
      _hostName = _ia.getHostName();
      String _key = _hostName + ":" + _portNum + random.nextInt();
      
      if (links.containsKey(_key)) {
	throw new TransportException(moduleName + "Link already exists to" +
				     "pooltcp://" + _key);
      }
      SocketChannel sc = SocketChannel.open();
      sc.configureBlocking(false);
      sc.connect(_ia);
      
      
      while (!sc.finishConnect()) {
	System.out.println(".");
      }
            
      PoolTCPLink _link = new PoolTCPLink(sc, this);
      _link.setLinkId(_key);
      
      links.put(_key, _link);
      transportHandler.registerLink(this, _link);
      _link.startLinkServices();

      registerChannel = true;
      poolSelector.wakeup();
      registerSocketChannel(sc, _link);
      registerChannel = false;


      return _link;
    } catch (Exception ioe) {
      String errorInfo = moduleName + " Unable to Setup Connection to [" + 
	_hostName + ":" + _portNum + "] => " + ioe;
      System.out.println(errorInfo);
      throw new TransportException(errorInfo);
    }
  }
  
  /** Set the interval for checking status of all links managed by this
      factory to the one specified in the argument */
  public void setLinkStatusInterval(long interval) {


  }
  
  /** Start communication services. In the case of PoolTCP for example, the
      thread which incorporates the serverSocket.accept() is started*/
  public boolean loadLinkFactoryServices() throws TransportException {
    /** This implies that there is no need to start up a PoolTCPServerSocket
	This is true in the case of clients, who don't anticipate connections
	from other clients e.g. JMS Clients*/
    if (pooltcpServerSocketPort == 0) {
      System.out.println(moduleName + "Will not be starting the ServerSocket"
			 + " Thread since the port specified is [0]");
    } else {
      try {
	ssc = ServerSocketChannel.open();
	ssc.socket().bind(new InetSocketAddress(pooltcpServerSocketPort));
	
	ssc.configureBlocking(false);
	ssc.register(poolSelector, SelectionKey.OP_ACCEPT);
	if (PoolTCPServerThread_Debug)
	  System.out.println(moduleName + 
			     "TCP ServerSocket created on port [" 
			     + pooltcpServerSocketPort + "]");
      } catch (Exception e) {
	String errorInfo = moduleName+ "Couldn't create TCPServerSocket on " +
	  "port [" + pooltcpServerSocketPort + "] " + e.toString();
	
	if (PoolTCPServerThread_Debug) {
	  System.out.println(errorInfo);
	  e.printStackTrace();
	}
	throw new TransportException(errorInfo);
      }/*end try-catch */
      
    } /*end if-else */
    start();
    System.out.println("Loaded PoolTCPLinkFactory communication services");
    return true;
  }
 
  /** Indicates if this factory can manage migration of links of a certain
      type */
  public boolean canManageLinkMigrationOfType(String linkType) {
    /**NBNI*/
    return false;
  }
  
  /** Used to migrate communications of a link to the one supported by
      links managed by this factory */
  public void manageLinkMigration(Link link) throws TransportException{
    /**NBNI*/
  }
  
  
  /** Sets the reference to the transport handler. This is done during the
      loadLinkFactory() operation within the transport handler 
      implementations. */
  public void setTransportHandler(TransportHandler transportHandler) {
    this.transportHandler = transportHandler;
  }
  
  /** This method is used to garbage collect any resources associated with
      the link factory */
  public void dispose() {
    Enumeration _links = links.elements();
    while(_links.hasMoreElements()) {
      PoolTCPLink _pooltcpLink = (PoolTCPLink) _links.nextElement();
      _pooltcpLink.closeLink();
    }
    
    if (pooltcpServerThread != null) {
      if (PoolTCPLinkFactory_Debug) {
	System.out.println(moduleName +"Disposing PoolTCPServerThread");
      }
      pooltcpServerThread.closeServerSocket();
      pooltcpServerThread = null;
    }
  }
  
  public void handleInboundPoolTCPLinkRequest(SocketChannel sc) {
    if (PoolTCPLinkFactory_Debug) {
      System.out.println(moduleName + "InboundConnection request received " +
			 sc);
    }
    Socket socket = sc.socket();
    String _hostName = socket.getInetAddress().getHostName();
    int _portNum = socket.getPort();
    String _linkId = _hostName + ":" + _portNum + random.nextInt();
    if (links.containsKey(_linkId)) {
      System.out.println(moduleName + "InboundConnection => " +
			 "Link aleady exists to " + _linkId);
    }
    try {
      PoolTCPLink _pooltcpLink = new PoolTCPLink(sc, this);
      _pooltcpLink.setLinkId(_linkId);
      links.put(_linkId, _pooltcpLink);
      transportHandler.registerLink(this, _pooltcpLink);
      _pooltcpLink.startLinkServices();
      
      registerSocketChannel(sc, _pooltcpLink);
    } catch (TransportException transEx) {
      System.out.println("moduleName" + transEx);
    } /*end try-catch */    
  }
  
  
  public void reportLinkLoss(Link linkLost) {
    String _linkId = linkLost.getLinkId();
    if (_linkId == null) {
      System.out.println(moduleName + "ReportLinkLoss Link Id is null");
      return;
    }
    if (links.containsKey(_linkId)) {
      transportHandler.manageLinkLoss((Link)links.remove(_linkId));
      System.out.println(moduleName + "Managed Link Loss");
      return;
    }
    
    System.out.println(moduleName +"Unaware of link [" + _linkId + 
		       "] whose connection loss is being reported");
  }
  
  
  public TransportHandler getTransportHandler() {
    if (transportHandler == null)
      System.out.println(moduleName + "TransportHandler reference is NULL");
    return transportHandler;
  }


  private void registerSocketChannel(SelectableChannel sc, 
				     PoolTCPLink poolTCPLink) {
    try {
      System.out.println(moduleName + 
			 "Registering SocketChannel to poolSelector");
      sc.register(poolSelector, SelectionKey.OP_READ);
      System.out.println(moduleName + 
			 "Registeration complete ...");
      poolLinks.put(sc, poolTCPLink);
      System.out.println(moduleName + poolSelector.keys());
    } catch (Exception e) {
      e.printStackTrace();
      System.out.println(moduleName + e); 
    }
  }

  public void run() {
    SelectionKey key=null;
    PoolTCPLink _poolTCPLink=null;
    while(keepLooping) {
      try {
	System.out.println(moduleName + "Waiting on poolSelector");
	int status = poolSelector.select();
	System.out.println(moduleName + "Returning from the poolSelector" + 
			   " selection process");
	
	while (registerChannel) {
	  sleep(100);
	}
	
	if (status == 0) {
	  System.out.println(moduleName + "poolSelector status returned=[0]");
	  continue;
	}
	Iterator it = poolSelector.selectedKeys().iterator();
	
	while(it.hasNext()) {
	  key = (SelectionKey) it.next();
	  ThreadPoolWorkerThread worker;
	  SelectableChannel _sc = key.channel();
	  _poolTCPLink = (PoolTCPLink) poolLinks.get(_sc);

	  if (key.isAcceptable()) {
	    System.out.println(moduleName +"ready to accept connections ");
	    SocketChannel sc = ssc.accept();
	    handleInboundPoolTCPLinkRequest(sc);
	  }
	  
	  if (key.isReadable()) {
	    System.out.println(moduleName +"ready to read!! ");
	    worker = threadPoolManager.getThreadPoolWorkerThread(_poolTCPLink);
	    
	    if (worker!= null) {
	      worker.performReadOperation(key);
	    } else {
	      System.out.println(moduleName +"Worker unavailable to read!!");
	    }
	    
	  }/** if key is readable */
	  
	  if (key.isWritable()) {
	    System.out.println(moduleName +"ready to write!! ");
	  }
	  it.remove(); /* remove key */
	} /*end while */
	
	
      } catch (Exception se) {
	se.printStackTrace();
	System.out.println(moduleName + se);
	if (key!=null && _poolTCPLink!=null) {
	  key.cancel();
	  _poolTCPLink.processLossOfSocketConnection();
	}
      } /*end try-catch */
    }/*end while(keepLooping) */
  }/*end method(run) */
  
  
  

  public static void main(String[] args) {
    if (args.length != 1) {
      System.out.println("Usage: java cgl.narada.transport.threadpool." +
			 "PoolTCPLinkFactory <portnum> ");
    }
    PoolTCPLinkFactory factory = new PoolTCPLinkFactory();
    Properties props = new Properties();
    props.put("PoolTCPServerPort", args[0]);
    factory.setLinkFactoryProperties(props);
    BufferedReader r = new BufferedReader(new InputStreamReader(System.in));
    String cmd;
    StringTokenizer tok;
    Hashtable links = new Hashtable();
    try {
      factory.loadLinkFactoryServices();      
      System.out.println("\nType h for usage options");
      while(true) {
	cmd = r.readLine();
	
	if (cmd==null) {
	  factory.dispose();
	  break;
	}

	if (cmd.startsWith("h")) {
	  System.out.println("cr <host> <port> \t \t \t Creates a link");
	  System.out.println("cl <link-id> \t \t \t Closes a link");
	  continue;
	}
	
	if (cmd.startsWith("cr")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 3) {
	    System.out.println("cr <host> <port> \t \t \t Creates a link");
	    continue;
	  }
	  Properties _props = new Properties();
	  tok.nextToken();
	  _props.put("hostname", tok.nextToken());
	  _props.put("portnum", tok.nextToken());
	  Link _link = factory.createLink(_props);
	  links.put(_link.getLinkId(), _link);
	  System.out.println("Created link with id " + 
			     _link.getLinkId());
	} /** if (creation of link */
	
	
	if (cmd.startsWith("cl")) {
	  tok = new StringTokenizer(cmd, " ");
	  if (tok.countTokens() != 2) {
	    System.out.println("cl <link-id> \t \t \t Closes a link");
	    continue;
	  }
	  tok.nextToken();
	  String _linkId = tok.nextToken();
	  if (links.containsKey(_linkId)) {
	    Link _link = (Link) links.remove(_linkId);
	    _link.closeLink();
	  } else {
	    System.out.println("Specified link Id is not correct!");
	  }
	} /* if (closing of a specified link ) */


	
	if (cmd.equalsIgnoreCase("q") ) {
	  factory.dispose();
	  break;
	}
      }/*end while(true)*/
    } catch (IOException ioe) {
      ioe.printStackTrace();
    } catch (TransportException transEx) {
      System.out.println(transEx);
    }/*end try-catch */
    System.out.println("PoolTCPLinkFactory: Exiting program ... end main()");
  }/*end-main*/
  
}

